package org.example.com.atguigu.day02;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.junit.Test;

import java.util.Arrays;

public class $01_RDDPartitions {

    /**
     * 通过集合创建RDD的分区数： sc.parallelize(集合[,numSlices])
     *      1、如果有设置numSlices参数,此时RDD的分区数 = 设置的numSlices参数
     *      2、如果没有设置numSlices参数,此时RDD的分区数 = defaultParallelism
     *         1、如果有在sparkconf中设置spark.default.parallelism参数,此时defaultParallelism = spark.default.parallelism参数值
     *          2、如果没有在sparkconf中设置spark.default.parallelism参数
     *              1、master=local, defaultParallelism = 1
     *              2、master=local[N], defaultParallelism = N
     *              3、master=local[*], defaultParallelism = 机器最大线程数
     *              4、master=yarn, defaultParallelism = math.max(所有executor总核数, 2)
     */
    @Test
    public void createRddByCollectionPartitions(){

        //1、创建JavaSparkcontext
        SparkConf conf = new SparkConf()
                .setMaster("local[4]")
                .setAppName("com.atguigu.day01.$01_RddCreate");
        //.set("spark.default.parallelism","6");
        JavaSparkContext sc = new JavaSparkContext(conf);

        //通过集合创建RDD
        //JavaRDD<Integer> rdd1 = sc.parallelize(Arrays.asList(1, 7, 5, 3, 2, 8),10);
        //TODO 通过集合创建RDD的时候,集合切片计划:
        //TODO     每个分区读取的集合元素角标范围[start,end): start = (分区号 * 集合长度 ) / 分区数  end=((分区号+1)*集合长度) /分区数
        JavaRDD<Integer> rdd1 = sc.parallelize(Arrays.asList(1, 7, 5, 3, 2, 8));

        System.out.println(rdd1.getNumPartitions());
    }

    /**
     * 通过读取文件创建RDD的分区数: sc.textFile(path[,minPartitions])
     *      1、如果有设置minPartitions参数,RDD的分区数>=minPartitions
     *      2、如果没有设置minPartitions参数，RDD的分区数>=math.min(defaultParallelism, 2)
     *      通过读取文件创建的RDD最终分区数由文件的切片数决定
     */
    @Test
    public void createRddByFilePartitions(){

        //1、创建JavaSparkcontext
        SparkConf conf = new SparkConf().setMaster("local[4]").setAppName("com.atguigu.day01.$01_RddCreate").set("spark.default.parallelism", "1");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<String> rdd1 = sc.textFile("datas/wc.txt",5);

        System.out.println(rdd1.getNumPartitions());
    }

    /**
     * 通过其他RDD衍生出的新RDD的分区数 = 父RDD的分区数
     */
    @Test
    public void createRddByRddPartitions(){
        //1、创建JavaSparkcontext
        SparkConf conf = new SparkConf().setMaster("local[4]").setAppName("com.atguigu.day01.$01_RddCreate");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<String> rdd1 = sc.textFile("datas/wc.txt");

        JavaRDD<String> rdd2 = rdd1.map(line -> {
            return line;
        });

        System.out.println(rdd2.getNumPartitions());
    }
}
